package net.wicp.tams.common.binlog.self.reader;

import java.io.File;
import java.io.IOException;
import java.util.Arrays;

import lombok.extern.slf4j.Slf4j;
import net.wicp.tams.common.Result;
import net.wicp.tams.common.apiext.ByteUtil;
import net.wicp.tams.common.apiext.StringUtil;
import net.wicp.tams.common.binlog.self.IBinlogRead;
import net.wicp.tams.common.binlog.self.IEventRead;
import net.wicp.tams.common.binlog.self.MysqlUtil;
import net.wicp.tams.common.binlog.self.bean.CountNum;
import net.wicp.tams.common.binlog.self.bean.EventBean;
import net.wicp.tams.common.binlog.self.bean.EventHeader;
import net.wicp.tams.common.binlog.self.bean.Host;
import net.wicp.tams.common.binlog.self.bean.Pos;
import net.wicp.tams.common.binlog.self.constant.BinLogVersion;
import net.wicp.tams.common.binlog.self.constant.Checksum;
import net.wicp.tams.common.binlog.self.constant.EventFlag;
import net.wicp.tams.common.binlog.self.constant.EventType;
import net.wicp.tams.common.binlog.self.event.AbsEvent;
import net.wicp.tams.common.binlog.self.event.FormatDescription;
import net.wicp.tams.common.binlog.self.event.GtidEvent;
import net.wicp.tams.common.binlog.self.event.rows.RowsEvent;
import net.wicp.tams.common.binlog.self.sender.ISender;
import net.wicp.tams.common.io.InputStreamRamdomRead;
import net.wicp.tams.common.thread.threadlocal.PerthreadManager;

/***
 * 读日志通过文件
 * 
 * @author zhoujunhui
 *
 */
@Slf4j
public class BinglogReadByFile extends InputStreamRamdomRead implements IBinlogRead {
	private EventBean firstEvent;
	private Checksum checksum;
	private final String fileName;
	private final ISender[] senders;
	private long curTime = 0;
	private long insertNum = 0;
	private long updateNum = 0;
	private long deleteNum = 0;
	private long endPos = -1;// 结束位置
	private String beginGtid;
	private String endGtid;

	public BinglogReadByFile(File file, Checksum checksum, ISender... senders) throws IOException {
		super(file);
		this.checksum = checksum;
		this.fileName = file.getName();
		if (senders != null && senders.length > 0) {// senders.length > 0不能
			this.senders = senders;
		} else {
			this.senders = null;
		}
		Host host = (Host) PerthreadManager.getInstance().createValue("zorro-host",Host.class).get(new Host());
		host.setHostIp("127.0.0.1");
		host.setPort(3306);
		host.setDefaultDb("mysql");
		host.setUser("root");
		host.setPwd("111111");
		PerthreadManager.getInstance().createValue("zorro-host",Host.class).set(host);
	}

	public BinglogReadByFile(File file, String dbPattern, String tbPattern, Checksum checksum, ISender... senders)
			throws IOException {
		this(file, checksum, senders);
		Host host = (Host) PerthreadManager.getInstance().createValue("zorro-host",Host.class).get(new Host());
		host.setDbPattern(dbPattern);
		host.setTbPattern(tbPattern);
	}

	@Override
	public Result checkHead() {
		try {
			this.seek(0);
		} catch (IOException e1) {
			log.error("不能跳到首位去检查，不合法", e1);
			throw new IllegalAccessError("不能跳到首位去检查，不合法");
		}
		try {
			final byte[] magic = this.readBytes(BINLOG_MAGIC.length);
			if (!Arrays.equals(magic, BINLOG_MAGIC)) {
				return Result.getError(String.format("不合法的binlog文件:[%s]", fileName));
			} else {
				firstEvent = readEvent();
				BinLogVersion binLogVersion = getVersion(firstEvent.getHead());
				if (binLogVersion == null || binLogVersion != BinLogVersion.v4) {
					return Result.getError(String.format("只支持v4的binlog，不支持的binlog版本:[%s]",
							binLogVersion == null ? "" : binLogVersion.name()));
				}
				return Result.getSuc();
			}
		} catch (Exception e) {
			return Result.getError(String.format("读binlog文件:[%s]错误", fileName));
		}
	}

	// 需要在第9个位开始读,读13位
	public BinLogVersion getVersion(EventHeader firstHead) {
		if (firstHead == null) {
			return null;
		}
		BinLogVersion retobj = null;
		switch (firstHead.getEventType()) {
		case FORMAT_DESCRIPTION_EVENT:
			retobj = BinLogVersion.v4;
			break;
		case START_EVENT_V3:
			long eventSize = firstHead.getEventSize();
			if (eventSize == (13 + 56)) {
				return BinLogVersion.v1;
			} else if (eventSize == (19 + 56)) {
				return BinLogVersion.v3;
			}
			break;
		default:
			break;
		}
		return retobj;
	}

	// private String gtids;

	private boolean isbegin = false;
	private String gtids;

	@Override
	public void read(long pos) throws IOException {
		Result checkRet = checkHead();
		if (!checkRet.isSuc()) {
			throw new RuntimeException(checkRet.getMessage());
		}
		// 初始化FormatDescription信息 firstEventHeader
		IEventRead reader = new FormatDescription(firstEvent);
		reader.parseBody(senders);
		if (pos > firstEvent.getHead().getNextEventPos()) {
			this.seek(pos);
		}
		while (this.available() > 0) {
			EventBean curEvent = readEvent();
			if (endPos > 0 && curEvent.getBeginHead() > endPos) {
				break;
			}
			if (curEvent.getHead().getEventType() == EventType.GTID_EVENT) {
				if (StringUtil.isNull(beginGtid) && StringUtil.isNull(endGtid)) {// 不以gtid为标准
					isbegin = true;
				}
				GtidEvent gtidEvent = (GtidEvent) curEvent.getEventRead();
				gtidEvent.parseBody(senders);
				this.gtids = gtidEvent.getGtid();
				if (StringUtil.isNotNull(beginGtid) && MysqlUtil.compare(gtids, beginGtid) >= 0) {
					isbegin = true;
				}
				if (StringUtil.isNotNull(endGtid) && MysqlUtil.compare(gtids, endGtid) > 0) {// 需要结束
					break;
				}
				continue;
			}
			if (!isbegin) {
				continue;
			}
			curTime = curEvent.getHead().getTimestamp();
			reader = curEvent.getEventRead();
			Result ret = null;
			if (reader instanceof RowsEvent) {
				ret = reader.parseBody(senders);
			} else {
				ret = reader.parseBody();
			}
			if (ret != null) {
				log.info(ret.getMessage());
			}
			// 统计增、删、改
			if (curEvent.getEventRead() instanceof RowsEvent) {
				RowsEvent rd = (RowsEvent) curEvent.getEventRead();
				switch (rd.getOptType()) {
				case insert:
					insertNum++;
					break;
				case update:
					updateNum++;
					break;
				case delete:
					deleteNum++;
					break;
				default:
					break;
				}
			}
		}
		log.info("日志[{}]解析完成,起始位置[{}]结束位置[{}]", fileName, pos, endPos);
	}

	private EventBean readEvent() throws IOException {
		if (this.available() <= 0) {
			return null;
		}
		long timestamp = ByteUtil.readLongL(readBytes(4));
		EventType eventType = EventType.get(readByte());
		long serverId = ByteUtil.readLongL(readBytes(4));
		long eventSize = ByteUtil.readLongL(readBytes(4));
		long nextEventPos = ByteUtil.readLongL(readBytes(4));
		int checkNum = checksum == null ? 0 : checksum.getByteNum();// 检验位数
		EventFlag eventFlag = EventFlag.get(ByteUtil.readIntL(readBytes(2)));
		byte[] body = readBytes((int) eventSize - 19 - checkNum);
		EventHeader head = EventHeader.builder().timestamp(timestamp).eventType(eventType).serverId(serverId)
				.eventSize(eventSize).nextEventPos(nextEventPos).eventFlag(eventFlag).build();
		EventBean retobj = new EventBean(head, body);
		retobj.setGtids(this.gtids);
		skip(checkNum);// 跳过
		log.info(
				"---------------------------事件[{}]开始位置：{}，数据开始位置{}----------------------------------------------------",
				eventType == null ? "" : eventType.name(), AbsEvent.row(retobj.getBeginHead()),
				AbsEvent.row(retobj.getBeginBody()));
		return retobj;
	}

	@Override
	public Pos curPos() {
		Pos ret = new Pos();
		ret.setFileName(fileName);
		ret.setPos(getCurPos());
		ret.setTime(curTime);
		return ret;
	}

	@Override
	public void resetNum() {
		insertNum = 0;
		updateNum = 0;
		deleteNum = 0;
	}

	@Override
	public CountNum getCountNum() {
		CountNum countNum = CountNum.builder().insertNum(insertNum).updateNum(updateNum).deleteNum(deleteNum).build();
		return countNum;
	}

	public long getEndPos() {
		return endPos;
	}

	public void setEndPos(long endPos) {
		this.endPos = endPos;
	}

	public void setBeginGtid(String beginGtid) {
		this.beginGtid = beginGtid;
	}

	public void setEndGtid(String endGtid) {
		this.endGtid = endGtid;
	}

}
